package org.joa.manager;

import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;

import org.joa.actor.Actor;
import org.joa.actor.ActorName;
import org.joa.actor.ActorProxy;
import org.joa.actor.ActorProxyInterface;
import org.joa.actor.Atomic;
import org.joa.actor.ReadOnly;
import org.joa.actor.exception.RemoteException;
import org.joa.actor.transaction.Transaction;
import org.joa.manager.exception.TransactionLockTimeoutException;
import org.joa.manager.persistence.JSONFilePersistenceManager;
import org.joa.manager.persistence.PersistenceManagerInterface;
import org.joa.util.Callback;

public class ActorManager implements ActorManagerInterface {
    public static long TRANSACTION_LOCK_TIMEOUT_MILLIS = 2 * 1000;

    public static ActorManager instance = new ActorManager();

    protected static class InvokeCallable implements Callable<Serializable> {
        private Method method;
        private Actor instance;
        private Object[] params;

        private Serializable result;

        private Logger logger = Logger.getLogger(InvokeCallable.class.getName());

        public InvokeCallable(Method method, Actor instance, Object[] params) {
            this.method = method;
            this.instance = instance;
            this.params = params;
        }

        @Override
        public Serializable call() throws Exception {
            logger.finer("call " + toString());

            try {
                this.result = (Serializable) method.invoke(instance, (Object[]) params);
            } catch (InvocationTargetException ex) {
                try {
                    throw ex.getTargetException();
                } catch (Exception e) {
                    throw e;
                } catch (Throwable e) {
                    throw ex;
                }
            }
            return result;
        }

        public Method getMethod() {
            return method;
        }

        public Actor getInstance() {
            return instance;
        }

        public Serializable getResult() {
            return result;
        }

        public String toString() {
            StringBuffer result = new StringBuffer();

            result.append(method.toString());

            if (method.isAnnotationPresent(Atomic.class)) {
                result.append(" {atomic}");
            }

            if (method.isAnnotationPresent(ReadOnly.class)) {
                result.append(" {read-only}");
            }

            return result.toString();
        }
    }

    protected Map<ActorName, ManagedActor> actors = new HashMap<ActorName, ManagedActor>();

    protected PersistenceManagerInterface persistenceManager = new JSONFilePersistenceManager(
        "data");

    protected AtomicLong currentRevision = new AtomicLong(1);

    private Logger logger = Logger.getLogger(ActorManager.class.getName());

    private ExecutorService executor = Executors.newCachedThreadPool();

    public ActorName create(Class<? extends Actor> actorClass) {
        try {
            ManagedActor managedActor = new ManagedActor(actorClass);
            actors.put(managedActor.getActorName(), managedActor);

            logger.fine("created actor " + managedActor.toString());

            commit(managedActor.getActorName(), currentRevision.get());

            return managedActor.getActorName();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void register(Actor actor) {
        ManagedActor managedActor = new ManagedActor(actor);
        actors.put(actor.self(), managedActor);

        logger.fine("registered actor " + managedActor.toString());
    }

    public Serializable createActorProxy(ActorName owner, Class<? extends Actor> actorClass,
        Class<?> interfaceClass) {
        ActorName remote = create(actorClass);
        return createActorProxy(remote, owner, interfaceClass);
    }

    public Serializable createActorProxy(ActorName remote, ActorName owner, Class<?> interfaceClass) {
        ActorProxy actorProxy = new ActorProxy(remote, owner, interfaceClass);
        return (Serializable) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class[] {
            Serializable.class, interfaceClass }, actorProxy);
    }

    public ManagedActor getActor(ActorName actorName) {
        ManagedActor actor = actors.get(actorName);

        if (actor == null) {
            throw new RuntimeException("An actor with name " + actorName + " does not exist");
        }
        return actor;
    }

    public long getCurrentRevision() {
        return currentRevision.get();
    }

    public void invoke(final ActorName actorName, final String methodName,
        final Serializable[] params, final Transaction transaction,
        final Callback<Serializable> callback) throws IOException, ClassNotFoundException,
        IllegalArgumentException, IllegalAccessException, InvocationTargetException {

        executor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Serializable result = invoke(actorName, methodName, params, transaction);
                    callback.onSuccess(result);
                } catch (Throwable ex) {
                    callback.onError(ex);
                }
            }
        });

    }

    public Serializable invoke(ActorName actorName, String methodName, Serializable[] params,
        Transaction transaction) throws IOException, ClassNotFoundException,
        IllegalArgumentException, IllegalAccessException, InvocationTargetException,
        RemoteException {

        ManagedActor managedActor = actors.get(actorName);

        InvokeCallable invokeCallable = getInvokeCallable(actorName, methodName, params);

        if (invokeCallable == null) {
            throw new RuntimeException("Method " + methodName + " not found for actor of type " +
                managedActor.getActorClass());
        }

        Method method = invokeCallable.getMethod();
        Actor actorInstance = invokeCallable.getInstance();

        if (transaction != null) {
            transaction = (Transaction) replaceActorProxies(transaction, actorName);
            beginTransaction(managedActor, transaction);
        }

        try {
            if (method.isAnnotationPresent(Atomic.class)) {
                transaction = actorInstance.executeAtomic(invokeCallable);
                return invokeCallable.getResult();
            } else {
                return invokeCallable.call();
            }
        } catch (RemoteException ex) {
            throw ex;
        } catch (Exception ex) {
            throw new RemoteException(ex);
        } finally {
            if (!method.isAnnotationPresent(ReadOnly.class) && transaction == null) {
                commit(actorName, currentRevision.get());
            }
        }
    }

    public InvokeCallable getInvokeCallable(ActorName actorName, String methodName,
        Serializable[] params) throws IOException, ClassNotFoundException,
        IllegalArgumentException, IllegalAccessException, InvocationTargetException,
        RemoteException {
        ManagedActor managedActor = actors.get(actorName);

        Class<?> actorClass = managedActor.getActorClass();
        Actor actorInstance = managedActor.getCurrentInstance();

        int paramsLength = (params != null) ? params.length : 0;

        replaceActorProxies(params, managedActor.getActorName());

        // Find matching method to invoke
        for (Method method : actorClass.getMethods()) {
            Class<?>[] methodParamTypes = method.getParameterTypes();

            if (method.getName().equals(methodName) && methodParamTypes.length == paramsLength) {
                boolean match = true;

                for (int idx = 0; idx < paramsLength; idx++) {
                    if (params[idx] == null) {
                        continue;
                    }

                    Class<?> paramClass = params[idx].getClass();

                    if (params[idx] instanceof ActorProxyInterface) {
                        ActorProxyInterface actorProxy = (ActorProxyInterface) params[idx];
                        paramClass = actorProxy.getProxyInterfaceClass();
                    }

                    Class<?> methodParamClass = methodParamTypes[idx];

                    if (!methodParamClass.equals(paramClass) &&
                        !paramClass.isInstance(methodParamClass)) {
                        match = false;
                        break;
                    }
                }

                if (match) {
                    return new InvokeCallable(method, actorInstance, (Object[]) params);
                }
            }
        }

        return null;
    }

    protected void beginTransaction(ManagedActor actor, Transaction transaction)
        throws TransactionLockTimeoutException {
        long startTime = System.currentTimeMillis();

        ActorName transactionActorName = transaction.getRemoteActorName();

        // TODO: Replace this polling loop with an asynchronous mechanism

        boolean waiting = false;
        do {
            synchronized (actor) {
                Transaction currentTransaction = actor.getCurrentTransaction();
                if (currentTransaction == null) {
                    actor.setCurrentTransaction(transaction);
                    logger.fine("Began transaction " + transaction + " on " + actor.toString());
                    return;
                } else if (currentTransaction.getRemoteActorName().equals(transactionActorName)) {
                    return;
                }
            }

            if (!waiting) {
                logger.fine("Waiting to begin transaction " + transaction + " on " +
                    actor.toString());

                waiting = true;
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        } while (System.currentTimeMillis() - startTime < TRANSACTION_LOCK_TIMEOUT_MILLIS);

        throw new TransactionLockTimeoutException(actor.getActorName(), transaction);
    }

    public void commit(ActorName actorName, long revision) throws IOException {
        updateCurrentRevision(revision);
        ManagedActor managedActor = actors.get(actorName);

        if (revision < managedActor.getCurrentRevision()) {
            throw new RuntimeException("Cannot commit to a prior revision");
        }

        persistenceManager.storeRevision(managedActor.getCurrentInstance(), revision);
        managedActor.setCurrentRevision(revision);
        managedActor.setCurrentTransaction(null);

        logger.info("Committed " + managedActor.toString());
    }

    public void rollback(ActorName actorName) throws RemoteException {
        ManagedActor managedActor = actors.get(actorName);

        synchronized (managedActor) {
            Actor currentInstance;
            try {
                currentInstance =
                    persistenceManager.restoreRevision(
                        managedActor.getActorName(),
                        managedActor.getActorClass(),
                        managedActor.getCurrentRevision());

                managedActor.setCurrentInstance(currentInstance);
                managedActor.setCurrentTransaction(null);
            } catch (Exception e) {
                throw new RemoteException(e);
            }
        }

        logger.info("rollback " + managedActor.toString());
    }

    public void reset() {
        actors.clear();
        currentRevision.set(1);
    }

    protected Object replaceActorProxies(Object obj, ActorName owner)
        throws IllegalArgumentException, IllegalAccessException {
        if (obj == null) {
            return null;
        } else if (obj instanceof ActorProxyInterface) {
            ActorProxyInterface actorProxy = (ActorProxyInterface) obj;
            return createActorProxy(
                actorProxy.getRemoteActorName(),
                owner,
                actorProxy.getProxyInterfaceClass());
        } else if (obj instanceof Object[]) {
            Object[] array = (Object[]) obj;
            for (int idx = 0; idx < array.length; idx++) {
                array[idx] = replaceActorProxies(array[idx], owner);
            }
        } else if (obj instanceof String) {
        } else {
            for (Field field : obj.getClass().getDeclaredFields()) {
                if (field.getType().isPrimitive() || Modifier.isFinal(field.getModifiers())) {
                    continue;
                }
                field.setAccessible(true);
                Object fieldValue = field.get(obj);
                field.set(obj, replaceActorProxies(fieldValue, owner));
            }
        }

        return obj;
    }

    protected void updateCurrentRevision(long revision) {
        synchronized (currentRevision) {
            if (currentRevision.get() < revision) {
                currentRevision.set(revision);
            }
        }
    }

    public void shutdown() {
        executor.shutdown();

        while (!executor.isTerminated()) {
            try {
                executor.awaitTermination(1, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
            }
        }
    }
}
